package com.atguigu.gmall.realtime.util;

import com.atguigu.gmall.realtime.commont.Constant;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @Author lzc
 * @Date 2023/4/19 15:22
 */
public class FlinkSourceUtil {
    
    public static KafkaSource<String> getKafkaSource(String groupId, String topic) {
        return KafkaSource.<String>builder()
            .setBootstrapServers(Constant.KAFKA_SERVERS)
            .setGroupId(groupId)
            .setTopics(topic)
            // 第一次消费的时候从指定的位置开始消费. 如果不是第一次,且状态中有消费记录, 则从状态中的消费记录继续消费
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                // 实现反序列化: 把字节数组变成字符串
                @Override
                public String deserialize(byte[] message) throws IOException {
                    if (message != null) {
                        return new String(message, StandardCharsets.UTF_8);
                    }
                    return null; // 如果这个地方返回的是 null, 流就会忽略这个值
                }
    
                // 永远返回 false: 表示流式一个无界流
                @Override
                public boolean isEndOfStream(String nextElement) {
                    return false;
                }
                // 定义反序列化之后的类型
                @Override
                public TypeInformation<String> getProducedType() {
//                    return Types.STRING;  // 当是基本类型,String, 元组的时候
//                    return TypeInformation.of(String.class);  // 当类型是普通的类(没有泛型)的时候
                    return TypeInformation.of(new TypeHint<String>() {});  // 当类型是一个带泛型的类型的时候
                }
            })  // 这个有 bug, 后面会出现
            // 只读取已提交的数据
            .setProperty("isolation.level", "read_committed")
            //.setClientIdPrefix("DimApp-")  // 当用一个消费者组消费多个 topic 的时候,有用
            .build();
    }
    
}
